package db.app

import com.cw.realtime.common.base.FlinkRunner.{SqlRunnerConfig, createBaseDic, createTopicDbTable, runSql}
import com.cw.realtime.common.constant.Constant.{TOPIC_DWD_TRADE_ORDER_DETAIL, TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS}
import com.cw.realtime.common.util.SqlUtil
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.{Table, TableResult}

object DwdTradeOrderSucDetail {


  def main(args: Array[String]): Unit = {
    implicit val config = SqlRunnerConfig(TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS, 10026)
    runSql { (_, tableEnv) =>
      createTopicDbTable(tableEnv)
      val paymentInfo = filterPaymentInfo(tableEnv)
      tableEnv.createTemporaryView("payment_info", paymentInfo)

      createOrderDetail(tableEnv)

      createBaseDic(tableEnv)

      val payOrder = intervalJoin(tableEnv)
      tableEnv.createTemporaryView("pay_order", payOrder)


      val joined = lookupJoin(tableEnv)

      createSink(tableEnv)

      joined.insertInto(TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS).execute()


    }


  }


  def createSink(tableEnv: StreamTableEnvironment): TableResult = {
    tableEnv.executeSql(
      s"""
create table $TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS (
  id string,
  user_id string,
  province_id string,
  order_id string,
  sku_id string,
  activity_id string,
  activity_rule_id string,
  coupon_id string,
  sku_name string,
  order_price string,
  sku_num string,
  payment_code string,
  payment_name string,
  split_total_amount string,
  split_activity_amount string,
  split_coupon_amount string,
  payment_time string,
  ts bigint,
  PRIMARY KEY (id) NOT ENFORCED


)  ${SqlUtil.getUpsertKafkaSinkSql(TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS)}

        """)
  }


  def lookupJoin(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
select
  id,
  user_id,
  province_id,
  order_id,
  sku_id,
  activity_id,
  activity_rule_id,
  coupon_id,
  sku_name,
  order_price,
  sku_num,
  payment_type payment_code,
  bd.info.dic_name payment_name,
  split_total_amount,
  split_activity_amount,
  split_coupon_amount,
  payment_time,
  ts
from pay_order po
join base_dic FOR SYSTEM_TIME AS OF po.proc_time AS  bd
  on po.payment_type = bd.rowkey

        """)
  }


  def intervalJoin(tableEnv: StreamTableEnvironment): Table = {
    tableEnv.sqlQuery(
      s"""
select
  od.id id,
  od.user_id,
  province_id,
  od.order_id,
  sku_id,
  activity_id,
  activity_rule_id,
  coupon_id,
  sku_name,
  order_price,
  sku_num,
  payment_type,
  split_total_amount,
  split_activity_amount,
  split_coupon_amount,
  callback_time payment_time,
  pi.ts ts,
  proc_time

from payment_info pi, $TOPIC_DWD_TRADE_ORDER_DETAIL od
WHERE pi.order_id = od.order_id
AND pi.raw_time BETWEEN od.raw_time - INTERVAL '15' MINUTE AND od.raw_time + INTERVAL '5' SECOND

        """)
  }


  def createOrderDetail(tableEnv: StreamTableEnvironment): TableResult = {
    tableEnv.executeSql(
      s"""
create table $TOPIC_DWD_TRADE_ORDER_DETAIL (
  id                string,
  user_id           string,
  province_id       string,
  order_id          string,
  sku_id            string,
  activity_id       string,
  activity_rule_id  string,
  coupon_id         string,
  sku_name          string,
  order_price       string,
  sku_num           string,
  create_time       string,
  split_total_amount        string,
  split_activity_amount     string,
  split_coupon_amount       string,
  ts                bigint,
  raw_time as TO_TIMESTAMP_LTZ(ts * 1000, 3),
  WATERMARK FOR raw_time AS raw_time - INTERVAL '5' SECOND

) ${SqlUtil.getKafkaSourceSql(TOPIC_DWD_TRADE_ORDER_DETAIL, TOPIC_DWD_TRADE_ORDER_DETAIL)}
          """)
  }


  def filterPaymentInfo(tableEnv: StreamTableEnvironment): Table = {
    tableEnv.sqlQuery(
      """
select
  `data`['id'] id,
  `data`['order_id'] order_id,
  `data`['user_id'] user_id,
  `data`['payment_type'] payment_type,
  `data`['total_amount'] total_amount,
  `data`['callback_time'] callback_time,
  ts,
  proc_time,
  raw_time

from topic_db
where
  `type` = 'update'
  and `database` = 'gmall'
  and `table` = 'payment_info'
  and `old`['payment_status']  is not null
  and `old`['payment_status'] = '1601'
  and `data`['payment_status'] = '1602'
        """)


  }


}
